package com.zhaosc.rx.example;

import org.junit.Test;

import com.zhaosc.rx.utils.SleepUtils;

import io.reactivex.Scheduler;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Action;
import io.reactivex.functions.Consumer;
import io.reactivex.schedulers.Schedulers;
import io.reactivex.subjects.AsyncSubject;
import io.reactivex.subjects.BehaviorSubject;
import io.reactivex.subjects.PublishSubject;
import io.reactivex.subjects.ReplaySubject;

public class SubjectExample {

	/**
	 * Observer会接收AsyncSubject的onComplete()之前的最后一个数据
	 */
	@Test
	public void testAsyncSubject() {

		AsyncSubject<String> subject = AsyncSubject.create();
		subject.onNext("helloworld1");
		subject.onNext("helloworld2");
		subject.onNext("helloworld3");

		// subject.onComplete();
		subject.subscribe(new Consumer<String>() {

			@Override
			public void accept(String t) throws Exception {
				System.out.println("*****" + t);
			}
		}, new Consumer<Throwable>() {

			@Override
			public void accept(Throwable t) throws Exception {
				t.printStackTrace();
			}
		}, new Action() {

			@Override
			public void run() throws Exception {
				System.out.println("****onComplete****");

			}
		});

		subject.onNext("helloworld4");
		subject.onNext("helloworld5");

		subject.onComplete();// 不调用此方法不发送数据

	}

	/**
	 * 发送订阅之前都一个数据，和订阅之后都全部数据
	 */
	@Test
	public void testBehaviorSubject() {
		BehaviorSubject<String> subject = BehaviorSubject.createDefault("default");

		// subject.onNext("hello1");
		subject.subscribe(new Consumer<String>() {

			@Override
			public void accept(String t) throws Exception {
				System.out.println("*****" + t);
			}
		}, new Consumer<Throwable>() {

			@Override
			public void accept(Throwable t) throws Exception {
				t.printStackTrace();
			}
		}, new Action() {

			@Override
			public void run() throws Exception {
				System.out.println("****onComplete****");

			}
		});

		subject.onNext("hello2");
		subject.onNext("hello3");

	}

	/**
	 * 不论什么时候订阅都发送全部数据
	 */
	@Test
	public void testReplaySubject() {
		// ReplaySubject<String> subject=ReplaySubject.createWithSize(2);

		ReplaySubject<String> subject = ReplaySubject.createWithSize(2);
		subject.onNext("0");
		subject.onNext("1");
		subject.onNext("2");
		subject.subscribe(new Consumer<String>() {

			@Override
			public void accept(String t) throws Exception {
				System.out.println("******consumer****" + t);
			}
		}, new Consumer<Throwable>() {

			@Override
			public void accept(Throwable t) throws Exception {
				System.out.println("Throwable");
			}
		}, new Action() {

			@Override
			public void run() throws Exception {
				System.out.println("*********onComplete************");
			}
		}, new Consumer<Disposable>() {

			@Override
			public void accept(Disposable t) throws Exception {
				System.out.println("********accept********");
			}
		});

		subject.onNext("3");
		subject.onNext("4");
	}

	/**
	 * 发送订阅之后的全部数据
	 */
	@Test
	public void testPublishSubject() {
		PublishSubject<String> subject = PublishSubject.create();
		subject.onNext("hello");
		subject.onNext("world");
		subject.subscribe(new Consumer<String>() {

			@Override
			public void accept(String t) throws Exception {
				System.out.println("******consumer****" + t);
			}
		}, new Consumer<Throwable>() {

			@Override
			public void accept(Throwable t) throws Exception {
				System.out.println("Throwable");
			}
		}, new Action() {

			@Override
			public void run() throws Exception {
				System.out.println("*********onComplete************");
			}
		}, new Consumer<Disposable>() {

			@Override
			public void accept(Disposable t) throws Exception {
				System.out.println("********accept********");
			}
		});

		// subject.onComplete();
		subject.onNext("woca");
		subject.onComplete();

	}
	/**
	 * 因为 subject发射元素的线程被指派到了 1/0线程，此时 I/O 线程正在初始化还没起来，
	 *  subject 发射前， Hello 和 world 这两个元素还在主线程中 ，而在从主线程往 1/0 线程转发的过程中，
	 *  由于 1/0 线程还没有起来，所以就被丢弃了。此时，无论 Thread 睡了多少秒， hello、 world 都不会被打 印出来 。
	 *  其实，解决方法很简单，使用 Observable.create()来替代subject，
	 *  它允许为每个订阅者精确 控制事件的发迭 ，这样就不会少打印 hello 和 world 了 。
	 */
	
	@Test
	public  void testMissing() {
		PublishSubject<String> subject = PublishSubject.create();
		subject.onNext("hello");
		subject.onNext("world");
		subject.subscribeOn(Schedulers.io())
		.subscribe(new Consumer<String>() {

			@Override
			public void accept(String t) throws Exception {
				System.out.println("******consumer****" + t);
			}
		}, new Consumer<Throwable>() {

			@Override
			public void accept(Throwable t) throws Exception {
				System.out.println("Throwable");
			}
		}, new Action() {

			@Override
			public void run() throws Exception {
				System.out.println("*********onComplete************");
			}
		}, new Consumer<Disposable>() {

			@Override
			public void accept(Disposable t) throws Exception {
				System.out.println("********accept********");
			}
		});

		subject.onComplete();
		subject.onNext("woca");
		//subject.onComplete();
		SleepUtils.sleep(1000L);
	}
	

}
